package com.consumer.consumer;

import com.consumer.model.RocketMqGroupTopic;
import com.consumer.service.MessageService;
import com.consumer.service.impl.MessageServiceImpl;
import com.consumer.utils.DateUtil;
import com.consumer.utils.JacksonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author Liunh
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = RocketMqGroupTopic.Group.TEST_GROUP,
        topic = RocketMqGroupTopic.Topic.TEST_TOPIC,
        selectorExpression = RocketMqGroupTopic.Tag.DEFAULT_TAG,
        consumeMode = ConsumeMode.CONCURRENTLY)
public class DemoMqConsumer implements RocketMQListener<String> {

    private final MessageService messageService;

    public DemoMqConsumer(MessageServiceImpl messageService) {
        this.messageService = messageService;
    }

    @Override
    public void onMessage(String json) {
        log.info(">>>>>> 消息中间件接收XXX数据：{}", json);
        try {
            if (StringUtils.isNotBlank(json)) {
                Object dataBase = JacksonUtil.jsonToBean(json, Object.class);
                if (dataBase != null) {
                    this.processHandle(dataBase);
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("消息中间件接收XXX数据异常！" + e.getMessage(), e);
        }
    }

    public void processHandle(Object dataBase) {
        final Date sendTime = DateUtil.getCurrentDate();
        final Integer dataType = 1;
        try {
            // todo

        } catch (Exception e) {
            throw new RuntimeException("消息中间件接收XXX数据异常！" + e.getMessage(), e);
        }
    }

}
